package org.budo.elasticsearch.jdbc.driver.statement;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import org.budo.elasticsearch.jdbc.driver.BudoElasticsearchJdbcConnection;
import org.budo.elasticsearch.jdbc.driver.BudoElasticsearchJdbcPreparedStatement;
import org.budo.elasticsearch.jdbc.driver.parser.jsql.JSqlParserStatement;
import org.budo.jdbc.driver.statement.AbstractJdbcStatement;
import org.budo.jdbc.driver.statement.JdbcStatement;
import org.budo.support.lang.util.StringUtil;
import org.elasticsearch.action.ActionFuture;
import org.elasticsearch.action.ActionRequestBuilder;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.sort.SortBuilder;
import org.elasticsearch.search.sort.SortBuilders;
import org.elasticsearch.search.sort.SortOrder;

import lombok.Getter;
import lombok.Setter;
import lombok.experimental.Accessors;
import lombok.extern.slf4j.Slf4j;
import net.sf.jsqlparser.expression.Expression;
import net.sf.jsqlparser.expression.JdbcParameter;
import net.sf.jsqlparser.expression.LongValue;
import net.sf.jsqlparser.expression.Parenthesis;
import net.sf.jsqlparser.expression.StringValue;
import net.sf.jsqlparser.expression.operators.conditional.AndExpression;
import net.sf.jsqlparser.expression.operators.conditional.OrExpression;
import net.sf.jsqlparser.expression.operators.relational.EqualsTo;
import net.sf.jsqlparser.expression.operators.relational.ExpressionList;
import net.sf.jsqlparser.expression.operators.relational.GreaterThan;
import net.sf.jsqlparser.expression.operators.relational.GreaterThanEquals;
import net.sf.jsqlparser.expression.operators.relational.InExpression;
import net.sf.jsqlparser.expression.operators.relational.MinorThan;
import net.sf.jsqlparser.expression.operators.relational.MinorThanEquals;
import net.sf.jsqlparser.expression.operators.relational.NotEqualsTo;
import net.sf.jsqlparser.schema.Column;
import net.sf.jsqlparser.statement.delete.Delete;
import net.sf.jsqlparser.statement.select.AllColumns;
import net.sf.jsqlparser.statement.select.Limit;
import net.sf.jsqlparser.statement.select.OrderByElement;
import net.sf.jsqlparser.statement.select.PlainSelect;
import net.sf.jsqlparser.statement.select.Select;
import net.sf.jsqlparser.statement.select.SelectExpressionItem;
import net.sf.jsqlparser.statement.select.SelectItem;
import net.sf.jsqlparser.statement.update.Update;

/**
 * @author limingwei
 */
@Accessors(fluent = true)
@Slf4j
@Getter
@Setter
public abstract class AbstractElasticsearchStatement extends AbstractJdbcStatement {
    /**
     * 表名
     */
    private String tableName;

    protected SearchHit[] executeSearchRequest(PreparedStatement preparedStatement, String[] selectColumnNames) throws SQLException {
        BudoElasticsearchJdbcConnection connection = (BudoElasticsearchJdbcConnection) preparedStatement.getConnection();
        TransportClient transportClient = connection.getTransportClient();

        String elasticsearchQuery = this.elasticsearchQuery(preparedStatement, selectColumnNames); // 查询条件

        SearchRequestBuilder searchRequestBuilder = transportClient.prepareSearch() //
                .setIndices(new String[] { connection.getCatalog() }) // 库
                .setTypes(this.tableName()) // 表
                // .setSearchType(SearchType.COUNT) //
                .setQuery(elasticsearchQuery); // 条件

        // 聚合
        // searchRequestBuilder.setAggregations("{}".getBytes());

        // 排序
        List<SortBuilder> sortBuilders = this.sortBuilders();
        for (SortBuilder sortBuilder : sortBuilders) {
            searchRequestBuilder.addSort(sortBuilder);
        }

        // 分页
        int[] limit = this.limit();

        searchRequestBuilder.setFrom(limit[0]) //
                .setSize(limit[1]);

        // 执行
        SearchResponse response = this.executeAndGet(searchRequestBuilder);

        JdbcStatement jdbcStatement = ((BudoElasticsearchJdbcPreparedStatement) preparedStatement).getJdbcStatement();
        log.info("#95 executeSearchRequest, jdbcStatement=" + jdbcStatement + ", elasticsearchQuery=" + elasticsearchQuery);

        return response.getHits().hits();
    }

    protected <T> T executeAndGet(ActionRequestBuilder actionRequestBuilder) throws SQLException {
        try {
            ActionFuture actionFuture = actionRequestBuilder.execute();
            T response = (T) actionFuture.actionGet();

            log.debug("#107 request={}, response={}", actionRequestBuilder, response);

            return response;
        } catch (Throwable e) {
            log.error("#107 executeAndGet, actionRequestBuilder=" + actionRequestBuilder, e);
            throw new SQLException(e);
        }
    }

    /**
     * 去掉列名的反引号
     */
    protected String columnToEsKey(Column column) {
        if (null == column) {
            return null;
        }

        String columnName = column.getColumnName();
        if (null == columnName || columnName.isEmpty()) {
            return columnName;
        }

        if (columnName.startsWith("`") && columnName.endsWith("`")) {
            return columnName.substring(1, columnName.length() - 1);
        }

        return columnName;
    }

    protected String elasticsearchQuery(PreparedStatement preparedStatement, String[] selectColumnNames) {
        String queryMatch = this.elasticsearchQueryMatch(preparedStatement, this.getWhere());

        // 查询条件
        String elasticsearchQuery = "{\"query\": ";
        if (null == queryMatch || queryMatch.isEmpty()) {
            elasticsearchQuery += " { \"match_all\" : { } } ";
        } else {
            elasticsearchQuery += queryMatch;
        }

        // 返回的列
        if (null != selectColumnNames) {
            elasticsearchQuery += " , ";
            String _source = StringUtil.join(selectColumnNames, ", ", "\"", "\"");
            elasticsearchQuery += "\"_source\": [ " + _source + " ] ";
        }

        return elasticsearchQuery + " } ";
    }

    /**
     * 排序
     */
    private List<SortBuilder> sortBuilders() {
        if (!(((JSqlParserStatement) this.getSqlStatement()).getStatement() instanceof Select)) {
            return Collections.EMPTY_LIST;
        }

        PlainSelect plainSelect = (PlainSelect) ((Select) ((JSqlParserStatement) this.getSqlStatement()).getStatement()).getSelectBody();
        List<OrderByElement> orderByElements = plainSelect.getOrderByElements();
        if (null == orderByElements || orderByElements.isEmpty()) {
            return Collections.EMPTY_LIST;
        }

        List<SortBuilder> list = new ArrayList<SortBuilder>();
        for (OrderByElement orderByElement : orderByElements) {
            Expression expression = orderByElement.getExpression();
            Column column = (Column) expression;

            String columnName = column.getColumnName();
            SortOrder order = orderByElement.isAsc() ? SortOrder.ASC : SortOrder.DESC;

            SortBuilder sortBuilder = SortBuilders.fieldSort(columnName).order(order);

            list.add(sortBuilder);
        }
        return list;
    }

    private int[] limit() {
        Limit limit = this.getLimit();
        if (null == limit) {
            return new int[] { -1, -1 }; // 全部
        }

        Expression offset = limit.getOffset();
        LongValue offsetValue = (LongValue) offset;

        Expression rowCount = limit.getRowCount();
        LongValue rowCountValue = (LongValue) rowCount;

        return new int[] { //
                null == offsetValue ? -1 : (int) offsetValue.getValue(), //
                null == rowCountValue ? -1 : (int) rowCountValue.getValue()//
        };
    }

    private String elasticsearchQueryMatch(PreparedStatement preparedStatement, Expression expression) {
        if (null == expression) {
            return null;
        }

        if (expression instanceof EqualsTo) {
            EqualsTo equalsTo = (EqualsTo) expression;
            Expression left = equalsTo.getLeftExpression();
            Expression right = equalsTo.getRightExpression();

            Column column = (Column) left;
            Object value = this.expressionValueWithQuot(preparedStatement, right);

            return " { \"match\" : " + " { \"" + column.getColumnName() + "\" : " + value + " } } ";
        }

        if (expression instanceof NotEqualsTo) {
            NotEqualsTo notEqualsTo = (NotEqualsTo) expression;
            Expression left = notEqualsTo.getLeftExpression();
            Expression right = notEqualsTo.getRightExpression();

            Column column = (Column) left;
            Object value = this.expressionValueWithQuot(preparedStatement, right);

            return " { \"bool\" : { \"must_not\" : " + " { \"match\" : { \"" + column.getColumnName() + "\" : " + value + " } } } } ";
        }

        if (expression instanceof AndExpression) {
            AndExpression andExpression = (AndExpression) expression;
            Expression left = andExpression.getLeftExpression();
            Expression right = andExpression.getRightExpression();
            return " { \"bool\" : { \"must\" : [ " + this.elasticsearchQueryMatch(preparedStatement, left) + ", " + this.elasticsearchQueryMatch(preparedStatement, right) + " ] } } ";
        }

        if (expression instanceof OrExpression) {
            OrExpression orExpression = (OrExpression) expression;
            Expression left = orExpression.getLeftExpression();
            Expression right = orExpression.getRightExpression();
            return " { \"bool\" : { \"should\" : [ " + this.elasticsearchQueryMatch(preparedStatement, left) + ", " + this.elasticsearchQueryMatch(preparedStatement, right) + " ] } } ";
        }

        if (expression instanceof GreaterThan) {
            GreaterThan greaterThan = (GreaterThan) expression;
            Expression left = greaterThan.getLeftExpression();
            Expression right = greaterThan.getRightExpression();

            Column column = (Column) left;
            LongValue longValue = (LongValue) right;

            return "{ \"range\" : { \"" + column.getColumnName() + "\" : { \"gt\" : " + longValue.getValue() + " } } } ";
        }

        if (expression instanceof GreaterThanEquals) {
            GreaterThanEquals greaterThanEquals = (GreaterThanEquals) expression;
            Expression left = greaterThanEquals.getLeftExpression();
            Expression right = greaterThanEquals.getRightExpression();

            Column column = (Column) left;
            LongValue longValue = (LongValue) right;

            return "{ \"range\" : { \"" + column.getColumnName() + "\" : { \"gte\" : " + longValue.getValue() + " } } } ";
        }

        if (expression instanceof MinorThan) {
            MinorThan minorThan = (MinorThan) expression;
            Expression left = minorThan.getLeftExpression();
            Expression right = minorThan.getRightExpression();

            Column column = (Column) left;
            LongValue longValue = (LongValue) right;

            return "{ \"range\" : { \"" + column.getColumnName() + "\" : { \"lt\" : " + longValue.getValue() + " } } } ";
        }

        if (expression instanceof MinorThanEquals) {
            MinorThanEquals minorThanEquals = (MinorThanEquals) expression;
            Expression left = minorThanEquals.getLeftExpression();
            Expression right = minorThanEquals.getRightExpression();

            Column column = (Column) left;
            LongValue longValue = (LongValue) right;

            return "{ \"range\" : { \"" + column.getColumnName() + "\" : { \"lte\" : " + longValue.getValue() + " } } } ";
        }

        if (expression instanceof InExpression) {
            InExpression inExpression = (InExpression) expression;

            Expression left = inExpression.getLeftExpression();
            Column column = (Column) left;

            ExpressionList expressionList = (ExpressionList) inExpression.getRightItemsList();
            List<Expression> expressions = expressionList.getExpressions();

            List<Object> inItems = new ArrayList<Object>();

            for (Expression each : expressions) {
                inItems.add(this.expressionValueWithQuot(preparedStatement, each));
            }

            String terms = " { \"terms\" : { \"" + column.getColumnName() + "\" : [ " + StringUtil.join(inItems) + " ] } } ";
            if (!inExpression.isNot()) { // IN ()
                return terms;
            }

            // NOT IN ()
            return " { \"bool\" : { \"must_not\" : " + terms + " } } ";
        }

        // 括号
        if (expression instanceof Parenthesis) {
            Parenthesis parenthesis = (Parenthesis) expression;
            return this.elasticsearchQueryMatch(preparedStatement, parenthesis.getExpression());
        }

        throw new RuntimeException("#133 where=" + expression + ", type" + expression.getClass());
    }

    /**
     * 需要的情况下带上单引号
     */
    protected String expressionValueWithQuot(PreparedStatement preparedStatement, Expression expression) {
        Object value = this.expressionValue(preparedStatement, expression);
        if (null == value) {
            return null;
        }

        if (value instanceof String || value instanceof Character) {
            return "\"" + value + "\"";
        }

        if (value instanceof Number || value instanceof Boolean) {
            return value.toString();
        }

        throw new IllegalArgumentException("#337 value.type=" + value.getClass() + ", preparedStatement=" + preparedStatement + ", expression=" + expression);
    }

    /**
     * 读取参数值
     */
    protected Object expressionValue(PreparedStatement preparedStatement, Expression expression) {
        if (expression instanceof LongValue) {
            LongValue longValue = (LongValue) expression;
            return longValue.getValue();
        }

        if (expression instanceof JdbcParameter) {
            BudoElasticsearchJdbcPreparedStatement statement = (BudoElasticsearchJdbcPreparedStatement) preparedStatement;
            JdbcParameter jdbcParameter = (JdbcParameter) expression;

            return statement.getParameterMap().get(jdbcParameter.getIndex());
        }

        if (expression instanceof StringValue) {
            StringValue stringValue = (StringValue) expression;
            return stringValue.getValue();
        }

        if ("false".equalsIgnoreCase(expression.toString())) {
            return false;
        }

        if ("true".equalsIgnoreCase(expression.toString())) {
            return true;
        }

        throw new IllegalArgumentException("#366 expression=" + expression + ", type=" + expression.getClass());
    }

    protected String[] getSelectItemColumnNames() {
        PlainSelect plainSelect = (PlainSelect) ((Select) ((JSqlParserStatement) this.getSqlStatement()).getStatement()).getSelectBody();
        List<SelectItem> selectItems = plainSelect.getSelectItems();

        List<String> selectColumnNames = new ArrayList<String>();

        for (SelectItem selectItem : selectItems) {
            if (selectItem instanceof AllColumns) {
                selectColumnNames.add("*");
                continue;
            }

            SelectExpressionItem selectExpressionItem = (SelectExpressionItem) selectItem;
            Expression expression = selectExpressionItem.getExpression();
            if (expression instanceof Column) {
                Column column = (Column) expression;
                String columnName = column.getColumnName();
                selectColumnNames.add(columnName);
                continue;
            }

            throw new IllegalArgumentException("#394 expression=" + expression + ", type=" + expression.getClass());
        }

        return selectColumnNames.toArray(new String[selectColumnNames.size()]);
    }

    private Limit getLimit() {
        if (((JSqlParserStatement) this.getSqlStatement()).getStatement() instanceof Select) {
            return ((PlainSelect) ((Select) ((JSqlParserStatement) this.getSqlStatement()).getStatement()).getSelectBody()).getLimit();
        }

        if (((JSqlParserStatement) this.getSqlStatement()).getStatement() instanceof Update) {
            return ((Update) ((JSqlParserStatement) this.getSqlStatement()).getStatement()).getLimit();
        }

        if (((JSqlParserStatement) this.getSqlStatement()).getStatement() instanceof Delete) {
            return ((Delete) ((JSqlParserStatement) this.getSqlStatement()).getStatement()).getLimit();
        }

        throw new IllegalArgumentException("#413 statement=" + ((JSqlParserStatement) this.getSqlStatement()).getStatement());
    }

    private Expression getWhere() {
        if (((JSqlParserStatement) this.getSqlStatement()).getStatement() instanceof Select) {
            return ((PlainSelect) ((Select) ((JSqlParserStatement) this.getSqlStatement()).getStatement()).getSelectBody()).getWhere();
        }

        if (((JSqlParserStatement) this.getSqlStatement()).getStatement() instanceof Update) {
            return ((Update) ((JSqlParserStatement) this.getSqlStatement()).getStatement()).getWhere();
        }

        if (((JSqlParserStatement) this.getSqlStatement()).getStatement() instanceof Delete) {
            return ((Delete) ((JSqlParserStatement) this.getSqlStatement()).getStatement()).getWhere();
        }

        throw new IllegalArgumentException("#429 statement=" + ((JSqlParserStatement) this.getSqlStatement()).getStatement());
    }
}